package com.bw.realtime_work_order.dws;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @author fang kuai
 * @date 2025/3/13 15:54
 */
public class DwsRefundOrder {
    public static void main(String[] args) {
        EnvironmentSettings build = EnvironmentSettings.newInstance().build();
        TableEnvironment tableEnv = TableEnvironment.create(build);
                            //创建统一 Kafka 源表
        tableEnv.executeSql("CREATE TABLE kafka_source (\n" +
                "  `database` STRING COMMENT '数据库名',\n" +
                "  `table` STRING COMMENT '表名',\n" +
                "  `type` STRING COMMENT '操作类型',\n" +
                "  `ts` BIGINT COMMENT 'Binlog时间戳',\n" +
                "  `xid` BIGINT COMMENT '事务ID',\n" +
                "  `commit` BOOLEAN COMMENT '事务提交标记',\n" +
                "  `data` RAW COMMENT '原始JSON数据',\n" +
                "  `event_time` AS TO_TIMESTAMP(FROM_UNIXTIME(`ts`)),\n" +
                "  WATERMARK FOR `event_time` AS `event_time` - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "     'connector' = 'kafka',\n" +
                "     'topic' = 'topic_db',\n" +
                "     'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
                "     'scan.startup.mode' = 'earliest-offset',\n" +
                "     'format' = 'json',\n" +
                "     'json.ignore-parse-errors' = 'true'\n" +
                " );");
        /*------------创建目标表结构------------*/
        //物流跟踪表
        tableEnv.executeSql("CREATE TABLE logistics_track_sink (\n" +
                "  id BIGINT,\n" +
                "  order_id BIGINT,\n" +
                "  logistics_status STRING,\n" +
                "  status_update_time TIMESTAMP(3),\n" +
                "  create_time TIMESTAMP(3)\n" +
                ") WITH (\n" +
                "  'connector' = 'jdbc',\n" +
                "  'url' = 'jdbc:mysql://hadoop102:3306/realtime_work_order',\n" +
                "  'table-name' = 'logistics_track',\n" +
                "  'username' = 'root',\n" +
                "  'password' = '123456'\n" +
                ");");
        //订单信息表
        tableEnv.executeSql("CREATE TABLE order_info_sink (\n" +
                "  order_id BIGINT,\n" +
                "  merchant_id INT,\n" +
                "  customer_id INT,\n" +
                "  order_amount DECIMAL(10,2),\n" +
                "  commit_ship_time TIMESTAMP(3),\n" +
                "  actual_ship_time TIMESTAMP(3),\n" +
                "  create_time TIMESTAMP(3),\n" +
                "  update_time TIMESTAMP(3)\n" +
                ") WITH (\n" +
                "  'connector' = 'jdbc',\n" +
                "  'url' = 'jdbc:mysql://mysql:3306/realtime_work_order',\n" +
                "  'table-name' = 'order_info',\n" +
                "  'username' = 'root',\n" +
                "  'password' = '123456'\n" +
                ");");
        //退款订单表
        tableEnv.executeSql("CREATE TABLE refund_order_sink (\n" +
                "  refund_order_id VARCHAR(20) PRIMARY KEY,\n" +
                "  order_id VARCHAR(20),\n" +
                "  customer_id VARCHAR(20),\n" +
                "  merchant_id VARCHAR(20),\n" +
                "  apply_refund_amount DECIMAL(10,2),\n" +
                "  refund_status VARCHAR(20),\n" +
                "  apply_time TIMESTAMP(3),\n" +
                "  finish_time TIMESTAMP(3),\n" +
                "  category_id VARCHAR(20),\n" +
                "  create_time TIMESTAMP(3)\n" +
                ") WITH (\n" +
                "  'connector' = 'jdbc',\n" +
                "  'url' = 'jdbc:mysql://hadoop102:3306/realtime_work_order',\n" +
                "  'table-name' = 'refund_order',\n" +
                "  'username' = 'root',\n" +
                "  'password' = '123456'\n" +
                ");");
               //----------实现动态分流---------//
        //物流跟踪数据处理
        tableEnv.executeSql("INSERT INTO logistics_track_sink\n" +
                "SELECT \n" +
                "  JSON_VALUE(`data`, '$.id' AS BIGINT) AS id,\n" +
                "  JSON_VALUE(`data`, '$.order_id' AS BIGINT) AS order_id,\n" +
                "  JSON_VALUE(`data`, '$.logistics_status') AS logistics_status,\n" +
                "  TO_TIMESTAMP(JSON_VALUE(`data`, '$.status_update_time')) AS status_update_time,\n" +
                "  TO_TIMESTAMP(JSON_VALUE(`data`, '$.create_time')) AS create_time\n" +
                "FROM unified_kafka_source\n" +
                "WHERE `database` = 'realtime_work_order' \n" +
                "  AND `table` = 'logistics_track'\n" +
                "  AND `type` = 'insert';");
        //订单信息数据处理
        tableEnv.executeSql("INSERT INTO order_info_sink\n" +
                "SELECT\n" +
                "  JSON_VALUE(`data`, '$.order_id' AS BIGINT) AS order_id,\n" +
                "  JSON_VALUE(`data`, '$.merchant_id' AS INT) AS merchant_id,\n" +
                "  JSON_VALUE(`data`, '$.customer_id' AS INT) AS customer_id,\n" +
                "  JSON_VALUE(`data`, '$.order_amount' AS DECIMAL(10,2)) AS order_amount,\n" +
                "  TO_TIMESTAMP(JSON_VALUE(`data`, '$.commit_ship_time')) AS commit_ship_time,\n" +
                "  TO_TIMESTAMP(JSON_VALUE(`data`, '$.actual_ship_time')) AS actual_ship_time,\n" +
                "  TO_TIMESTAMP(JSON_VALUE(`data`, '$.create_time')) AS create_time,\n" +
                "  TO_TIMESTAMP(JSON_VALUE(`data`, '$.update_time')) AS update_time\n" +
                "FROM unified_kafka_source\n" +
                "WHERE `database` = 'realtime_work_order'\n" +
                "  AND `table` = 'order_info'\n" +
                "  AND `type` = 'insert';");
        //退款订单数据处理
        tableEnv.executeSql("INSERT INTO refund_order_sink\n" +
                "SELECT\n" +
                "  JSON_VALUE(`data`, '$.refund_order_id') AS refund_order_id,\n" +
                "  JSON_VALUE(`data`, '$.order_id') AS order_id,\n" +
                "  JSON_VALUE(`data`, '$.customer_id') AS customer_id,\n" +
                "  JSON_VALUE(`data`, '$.merchant_id') AS merchant_id,\n" +
                "  JSON_VALUE(`data`, '$.apply_refund_amount' AS DECIMAL(10,2)) AS apply_refund_amount,\n" +
                "  JSON_VALUE(`data`, '$.refund_status') AS refund_status,\n" +
                "  TO_TIMESTAMP(JSON_VALUE(`data`, '$.apply_time')) AS apply_time,\n" +
                "  TO_TIMESTAMP(JSON_VALUE(`data`, '$.finish_time')) AS finish_time,\n" +
                "  JSON_VALUE(`data`, '$.category_id') AS category_id,\n" +
                "  TO_TIMESTAMP(JSON_VALUE(`data`, '$.create_time')) AS create_time\n" +
                "FROM unified_kafka_source\n" +
                "WHERE `database` = 'realtime_work_order'\n" +
                "  AND `table` = 'refund_order'\n" +
                "  AND `type` = 'insert';");
        tableEnv.executeSql("select * from refund_order_sink").print();



       // 发货风险订单（承诺时间到但未发货）



    }
}
